package com.atguigu.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/2/5
 */
public class Demo9_DefineTime
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        /*
                通过sql语句定义一张表，提供时间属性的列
                        字段名 AS PROCTIME()： 定义处理时间

                        定义事件时间:
                            事件时间列，必须是 TIMESTAMP  类型 或 TIMESTAMP_LTZ 类型
                                事件时间从数据的某个时间属性中提取。
                                    原始数据的时间属性是  2020-04-15 20:13:40.564 ，要使用 TIMESTAMP(3)
                                    原始数据的时间属性是  2020-04-15 20:13:40     ，要使用 TIMESTAMP(0)

                                    原始数据是毫秒的时间戳，1618989564564 使用  TO_TIMESTAMP_LTZ(ts, 3)
                                    原始数据是秒的时间戳，1618989564     使用  TO_TIMESTAMP_LTZ(ts, 0)

                        还需声明水印策略。
         */
        String createTableSql = "create table t1( id string,ts bigint ,vc int," +
            "                                       pt AS PROCTIME() ," +
            "                                       et AS TO_TIMESTAMP_LTZ(ts, 3) ," +
            "                                       WATERMARK FOR et AS et - INTERVAL '0.001' SECOND ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        tableEnvironment.executeSql(createTableSql);

        tableEnvironment.sqlQuery("select * from t1")
                        .execute()
                        .print();

    }
}
